package com.ada.spark.practice

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object Practice01 {

    def main(args: Array[String]): Unit = {
        //创建配置对象
        val conf: SparkConf = new SparkConf().setAppName("JsonTest").setMaster("local[*]")
        //创建SparkSql的环境对象
        val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
        //导入隐式转换函数
        import spark.implicits._

        //加载数据
        val tbStockRdd = spark.sparkContext.textFile("in/tbStock.txt")
        val tbStockDS = tbStockRdd.map(_.split(",")).map(attr => tbStock(attr(0), attr(1), attr(2))).toDS
        //tbStockDS.show()

        val tbStockDetailRdd = spark.sparkContext.textFile("in/tbStockDetail.txt")
        val tbStockDetailDS = tbStockDetailRdd.map(_.split(",")).map(attr => tbStockDetail(attr(0), attr(1).trim().toInt, attr(2), attr(3).trim().toInt, attr(4).trim().toDouble, attr(5).trim().toDouble)).toDS
        //tbStockDetailDS.show()

        val tbDateRdd = spark.sparkContext.textFile("in/tbDate.txt")
        val tbDateDS = tbDateRdd.map(_.split(",")).map(attr => tbDate(attr(0), attr(1).trim().toInt, attr(2).trim().toInt, attr(3).trim().toInt, attr(4).trim().toInt, attr(5).trim().toInt, attr(6).trim().toInt, attr(7).trim().toInt, attr(8).trim().toInt, attr(9).trim().toInt)).toDS
        //tbDateDS.show()

        //注册表
        tbStockDS.createOrReplaceTempView("tbStock")
        tbDateDS.createOrReplaceTempView("tbDate")
        tbStockDetailDS.createOrReplaceTempView("tbStockDetail")

        spark.sql("select * from tbStock").show()
        spark.sql("select * from tbStockDetail").show()
        spark.sql("select * from tbDate").show()

        //1、统计所有订单中每年的销售单数、销售总额
        spark.sql(
            """
              |SELECT c.theyear, COUNT(DISTINCT a.ordernumber), SUM(b.amount)
              |FROM tbStock a
              |JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |JOIN tbDate c ON a.dateid = c.dateid
              |GROUP BY c.theyear
              |ORDER BY c.theyear
            """.stripMargin).show()

        //2、统计每年，每个订单一共有多少销售额
        spark.sql(
            """
              |SELECT a.dateid, a.ordernumber, SUM(b.amount) AS SumOfAmount
              |FROM tbStock a
              |JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |GROUP BY a.dateid, a.ordernumber
            """.stripMargin).show()

        //3、 以上一步查询结果为基础表，和表tbDate使用dateid join，求出每年最大金额订单的销售额
        spark.sql(
            """
              |SELECT theyear, MAX(c.SumOfAmount) AS SumOfAmount
              |FROM (SELECT a.dateid, a.ordernumber, SUM(b.amount) AS SumOfAmount
              |	FROM tbStock a
              |		JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |	GROUP BY a.dateid, a.ordernumber
              |	) c
              |	JOIN tbDate d ON c.dateid = d.dateid
              |GROUP BY theyear
              |ORDER BY theyear DESC
            """.stripMargin).show()
        //4、 求出每年每个货品的销售额
        spark.sql(
            """
              |SELECT c.theyear, b.itemid, SUM(b.amount) AS SumOfAmount
              |FROM tbStock a
              |JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |JOIN tbDate c ON a.dateid = c.dateid
              |GROUP BY c.theyear, b.itemid
            """.stripMargin).show()

        //5、在第一步的基础上，统计每年单个货品中的最大金额
        spark.sql(
            """
              |SELECT d.theyear, MAX(d.SumOfAmount) AS MaxOfAmount
              |FROM (SELECT c.theyear, b.itemid, SUM(b.amount) AS SumOfAmount
              |	FROM tbStock a
              |		JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |		JOIN tbDate c ON a.dateid = c.dateid
              |	GROUP BY c.theyear, b.itemid
              |	) d
              |GROUP BY d.theyear
            """.stripMargin).show()

        //6、用最大销售额和统计好的每个货品的销售额join，以及用年join，集合得到最畅销货品那一行信息
        spark.sql(
            """
              |SELECT DISTINCT e.theyear, e.itemid, f.MaxOfAmount
              |FROM (SELECT c.theyear, b.itemid, SUM(b.amount) AS SumOfAmount
              |	FROM tbStock a
              |		JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |		JOIN tbDate c ON a.dateid = c.dateid
              |	GROUP BY c.theyear, b.itemid
              |	) e
              |	JOIN (SELECT d.theyear, MAX(d.SumOfAmount) AS MaxOfAmount
              |		FROM (SELECT c.theyear, b.itemid, SUM(b.amount) AS SumOfAmount
              |			FROM tbStock a
              |				JOIN tbStockDetail b ON a.ordernumber = b.ordernumber
              |				JOIN tbDate c ON a.dateid = c.dateid
              |			GROUP BY c.theyear, b.itemid
              |			) d
              |		GROUP BY d.theyear
              |		) f ON e.theyear = f.theyear
              |		AND e.SumOfAmount = f.MaxOfAmount
              |ORDER BY e.theyear
            """.stripMargin).show()
    }
}

case class tbStock(ordernumber: String, locationid: String, dateid: String) extends Serializable

case class tbStockDetail(ordernumber: String, rownum: Int, itemid: String, number: Int, price: Double, amount: Double) extends Serializable

case class tbDate(dateid: String, years: Int, theyear: Int, month: Int, day: Int, weekday: Int, week: Int, quarter: Int, period: Int, halfmonth: Int) extends Serializable